package com.rabbit.producer.broker;

import com.rabbit.api.Message;
import com.rabbit.api.MessageType;
import com.rabbit.producer.constant.BrokerMessageConst;
import com.rabbit.producer.constant.BrokerMessageStatus;
import com.rabbit.producer.entity.BrokerMessage;
import com.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * 真正发送不同类型的消息实现类
 * @author: 顾志杰
 */
@Slf4j
@Component
public class RabbitBrokerImpl implements RabbitBroker {

    @Autowired
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Autowired
    private MessageStoreService messageStoreService;

    /**
     * 迅速发消息
     * @param message
     */
    @Override
    public void rapidSend(Message message) {
        message.setMessageType(MessageType.RAPID);
        sendKernel(message);
    }
    /**
     * 确认消息
     * @param message
     */
    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        sendKernel(message);
    }
    /**
     * 可靠消息
     * @param message
     */
    @Override
    public void reliantSend(Message message) {
        message.setMessageType(MessageType.RELIANT);

        BrokerMessage bm = messageStoreService.selectByPrimaryKey(message.getMessageId());
        if(bm == null){
            /*1.在数据库创建发送日志*/
            Date date = new Date();
            BrokerMessage brokerMessage = new BrokerMessage();
            brokerMessage.setMessageId(message.getMessageId());
            brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());
            /*tryCount最开始创建的时候不需要进行设置*/
            /*最大容忍的发送时间是当前时间加上一分钟*/
            brokerMessage.setNextRetry(DateUtils.addMinutes(date, BrokerMessageConst.TIMEOUT));
            brokerMessage.setCreateTime(date);
            brokerMessage.setUpdateTime(date);
            brokerMessage.setMessage(message);
            messageStoreService.insert(brokerMessage);
        }
        /*2.发送消息*/
        sendKernel(message);
    }

    @Override
    public void sendMessages() {
        List<Message> messages = MessageHolder.clear();
        messages.forEach(message -> {
            MessageHolderAsyncQueue.submit((Runnable)()->{
                String routingKey = message.getRoutingKey();
                String topic = message.getTopic();
                CorrelationData correlationData = new CorrelationData(
                        String.format("%s#%s#%s",message.getMessageId()
                                ,System.currentTimeMillis(),
                                message.getMessageType()));
                RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
                rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);
            });
        });
        log.info("#RabbitBrokerImpl.sendMessages# send to rabbitMQ size : {}",messages.size());
    }

    /**
     * 发送消息的核心方法 使用异步线程池进行发送消息
     * @param message
     */
    private void sendKernel(Message message){

        AsyncBaseQueue.submit((Runnable)()->{
            String routingKey = message.getRoutingKey();
            String topic = message.getTopic();
            CorrelationData correlationData = new CorrelationData(
                    String.format("%s#%s#%s",message.getMessageId()
                            ,System.currentTimeMillis(),
                            message.getMessageType()));
            RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
            rabbitTemplate.convertAndSend(topic,routingKey,message,correlationData);
            log.info("#RabbitBrokerImpl.sendKernel# send to rabbitMQ messageId : {}",message.getMessageId());
        });

    }
}
